package com.shujia.sql

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SaveMode, SparkSession}

object Demo2SourceAPI {
  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession
      .builder()
      .appName("Demo2SourceAPI")
      .master("local")
      .getOrCreate()

    /**
     * 1、读取文本类的数据，需要手动加上列名
     */
    val stuDF: DataFrame =
      spark.read
        .format("csv")
        .option("sep", ",") // 将分隔符指定为 ,
        .schema("id Int,name String,age Int,gender String,clazz String") // 手动指定每一列数据的列名及类型
        .load("Spark/data/stu/students.txt")

    stuDF.show()

    /**
     * 2、读取json格式的数据
     * json格式自带结构，所以不需要指定schema
     * 但是列名被存储了多次，会造成空间上浪费
     */
    val stuJSONDF: DataFrame =
      spark.read
        .format("json")
        .load("Spark/data/stu/students.json")

    stuJSONDF.show()
    //    // 通过DF生成一个parquet文件
    //    stuJSONDF
    //      .write
    //      // 指定写入的模式: Append追加、Overwrite覆盖、ErrorIfExists存在就报错、Ignore存在即忽略
    //      .mode(SaveMode.Overwrite)
    //      .format("parquet") // 指定输出的格式
    //      .save("Spark/data/stu/parquet")

    /**
     * 3、读取parquet文件
     * 自带表结构，默认会使用snappy对数据进行压缩
     */

    val stuParquetDF: DataFrame =
      spark.read
        .format("parquet")
        .load("Spark/data/stu/parquet")

    stuParquetDF.show()


    //    // 通过DF生成一个orc文件
    //    stuJSONDF
    //      .write
    //      // 指定写入的模式: Append追加、Overwrite覆盖、ErrorIfExists存在就报错、Ignore存在即忽略
    //      .mode(SaveMode.Overwrite)
    //      .format("orc") // 指定输出的格式
    //      .save("Spark/data/stu/orc")

    /**
     * 4、读取orc文件
     * 自带表结构，默认会使用snappy对数据进行压缩，压缩率最高
     */

    val stuORCDF: DataFrame =
      spark.read
        .format("orc")
        .load("Spark/data/stu/orc")

    stuORCDF.show()

    /**
     * 5、通过JDBC读取MySQL的数据
     */

    val jdbcDF: DataFrame = spark.read
      .format("jdbc")
      .option("url", "jdbc:mysql://master:3306/student")
      .option("dbtable", "student")
      .option("user", "root")
      .option("password", "123456")
      .load()

    jdbcDF.show()

    /**
     * 6、通过RDD创建DataFrame
     */
    //导入隐式转换
    import spark.implicits._

    val sc: SparkContext = spark.sparkContext
    val stuRDD: RDD[String] = sc.textFile("Spark/data/stu/students.txt")

    val df: DataFrame = stuRDD.map(line => {
      val splits: Array[String] = line.split(",")
      val id: Int = splits(0).toInt
      val name: String = splits(1)
      val age: Int = splits(2).toInt
      val gender: String = splits(3)
      val clazz: String = splits(4)
      //      (id, name, age, gender, clazz)
      Students(id, name, age, gender, clazz)
    })
      //      .toDF("id", "name", "age", "gender", "clazz")
      .toDF()

    df.show()

    // DataFrame to RDD
    val rdd: RDD[Row] = df.rdd


  }

}

case class Students(id: Int, name: String, age: Int, gender: String, clazz: String)
